
/*
    消息管理模块设计
        1，消息的属性：消息ID，消息是否持久化标志，消息的rounting_key（匹配字符串）
        2，消息持久化功能的实现：将消息保存在文件中，保存的内容：消息的属性，消息的内容，以及消息是否有效，这部分相当于持久化磁盘存储
        3，消息结构的所有成员 ：以上内容再加上消息的起始位置（相对于文件开始位置的偏移量），还有消息的长度，这部分相等于内存存储

    消息的持久化管理：
    1，以队列为单元进行消息的持久化存储：如果使用一个文件来存储所有队列的消息，
         那么会存在同时访问不同的队列的线程安全问题，所以需要加锁，加锁之后，就会存在频繁的锁竞争，效率太低。
            所以使用每个队列对应一个文件，来存储消息，就不会冲突了。
    2，存储时的格式规定：4字节长度|数据|4字节长度|数据……（解决粘包问题）
*/
#ifndef __M_MESSAGE_H__
#define __M_MESSAGE_H__
#include "../mqcommon/helpper.hpp"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/mq_message.pb.h" //消息的结构化描述在该部分使用protobuf生成
#include <iostream>
#include <string>
#include <memory>
#include <list>
#include <mutex>
#include <unordered_map>

namespace xgmq
{
    using MessagePtr = std::shared_ptr<Message>;
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"
#define MESSAGE_VALID "1"   // 消息有效
#define MESSAGE_INVALID "0" // 消息无效
    // 消息持久化功能的实现（文件存储）
    // 消息将来是要传递给某个队列的，消息的载体是队列
    class MessageMapper
    {
    public:
        MessageMapper(std::string &basedir, const std::string &qname)
            : _qname(qname)
        {
            if (basedir.back() != '/')
                basedir.push_back('/');
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;
            // 创建basedir目录
            assert(FileHelper::createDirector(basedir));

            if (FileHelper(_datafile).isExits() == false)
            {
                FileHelper::createFile(_datafile);
            }
        }
        // 创建文件，只需创建保存消息的文件即可
        bool cresteFile()
        {
            bool ret = FileHelper::createFile(_datafile);
            if (ret == false)
            {
                DLOG("创建文件失败\n");
                return false;
            }
            return true;
        }
        // 移除文件（两个都移除）
        void rmoveFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }
        // 新增一条消息
        bool insert(MessagePtr &msg)
        {
            return insert(_datafile, msg);
        }
        // 移除一条消息，只需将这条消息的有效标志位置为"0"即可
        bool remove(MessagePtr &msg)
        {
            // std::string body1=msg->payload().SerializeAsString();
            // DLOG("修改标志位之前序列化的长度:%ld\n",body1.size());//71,由于在插入消息的时候位设置有效位导致长度不一致
            // 将标志位设置为无效
            msg->mutable_payload()->set_valid(MESSAGE_INVALID);
            // 将该消息进行序列化
            std::string body = msg->payload().SerializeAsString();
            // DLOG("修改标志位之后序列化的长度:%ld\n",body.size());//74
            // DLOG("msg->length():%d",msg->length());//71
            if (body.size() != msg->length())
            {
                DLOG("无法写入，会覆盖其他消息...\n");
                return false;
            }
            // 获取这条消息的起始位置
            size_t offset = msg->offset();
            // 从该位置开始写入，覆盖掉之前的
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), offset, body.size());
            if (ret == false)
            {
                DLOG("移除一条消息失败...\n");
                return false;
            }
            return true;
        }
        // 进行数据备份和垃圾回收
        // 这里使用链表来保存消息对象
        std::list<MessagePtr> gc()
        {
            std::list<MessagePtr> result;
            bool ret;
            ret = load(result);
            // DLOG("垃圾回收时,得到的有效消息数量:%ld", result.size());
            if (ret == false)
            {
                DLOG("加载文件消息出现问题...\n");
                return result;
            }
            // 2，将有效消息，进行序列化存储到临时文件中
            if (FileHelper(_tmpfile).isExits() == false)
            {
                FileHelper::createFile(_tmpfile);
            }
            for (auto &msg : result)
            {
                // DLOG("向临时文件写入数据:%s",msg->payload().body().c_str());
                ret = insert(_tmpfile, msg);
                if (ret = false)
                {
                    DLOG("向临时文件写入出错...\n");
                    return result;
                }
            }
            // DLOG("垃圾回收后,向临时文件写入的数据大小为 :%ld", FileHelper(_tmpfile).size());
            // 3，将源文件删除
            FileHelper::removeFile(_datafile);
            // 4，将临时文件重命名为源文件
            ret = FileHelper(_tmpfile).rename(_datafile);

            return result;
        }

    private:
        bool load(std::list<MessagePtr> &result)
        {
            // 1，读取文件中的消息
            // 消息在文件中的存储格式为 4字节长度|数据|4字节长度|数据....
            FileHelper helper(_datafile);
            size_t msg_size; // 用来获取消息的大小（也就是消息的前4个字节）
            size_t offset = 0;
            if (helper.isExits() == false)
            {
                DLOG("文件不存在%s...", _datafile.c_str());
                helper.createFile(_datafile);
            }
            size_t fsize = helper.size(); // 文件大小竟然为0(由于重复创建就文件导致)
            bool ret;
            while (offset < fsize)
            {
                // 将前4个字节放到msg_size中
                helper.read((char *)&msg_size, offset, sizeof(size_t));
                offset += sizeof(size_t);

                std::string body(msg_size, '\0'); // 开空间保存数据
                ret = helper.read(&body[0], offset, msg_size);
                if (ret == false)
                {
                    DLOG("读取消息数据失败....\n");
                    return false;
                }
                MessagePtr msgp = std::make_shared<Message>();
                msgp->mutable_payload()->ParseFromString(body); // 反序列化
                msgp->set_offset(offset);
                offset += msg_size;

                if (msgp->payload().valid() == MESSAGE_INVALID)
                {
                    // DLOG("该消息无效,无法添加到内存中...\n");
                    continue; // 该消息无效
                }
                result.push_back(msgp);
            }

            return true;
        }
        bool insert(const std::string &filename, MessagePtr &msg)
        {
            // 1，将这条消息进行序列化
            std::string body = msg->payload().SerializeAsString();
            // 2，获取文件大小，写入文件后面
            FileHelper helper(filename);
            if (helper.isExits() == false)
            {
                FileHelper::createFile(filename);
            }
            size_t fsize = helper.size(); // 文件大小
            // 写入数据大小
            size_t len = body.size();
            bool ret = helper.write((char *)&len, fsize, sizeof(size_t));
            if (ret == false)
            {
                DLOG("写入数据长度失败...\n");
            }
            // 写入数据
            ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
            if (ret == false)
            {
                DLOG("写入数据内容失败...\n");
                return false;
            }
            msg->set_offset(fsize+sizeof(size_t));
            msg->set_length(body.size());
            return true;
        }

    private:
        std::string _qname;    // 队列名称
        std::string _datafile; // 保存消息的文件（实现持久化存储的文件）
        std::string _tmpfile;  // 临时文件，在进行垃圾回收时用来代替源文件的
    };

    // 队列中的消息
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string &basedir, const std::string &qname)
            : _mapper(basedir, qname),
              _qname(qname)
        {
            _total_count = _valid_count = 0;
        }
        // 恢复历史数据
        void recovery()
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            std::list<MessagePtr> msgs = _mapper.gc();
            for (auto &msg : msgs)
            {
                if (msg->payload().properties().deliver_mode() == DeliveryMode::DURABLE)
                    _durable_msg.insert(std::make_pair(msg->payload().properties().id(), msg));

                _send_msgs.push_back(msg);
            }
            _total_count = _valid_count = msgs.size();
        }
        // 新增一条消息
        bool insert(const basicProperties *pb, const std::string &body, bool queue_is_durable)
        {
            // 1,构造一个消息对象
            MessagePtr msp = std::make_shared<Message>();
            msp->mutable_payload()->set_body(body);
            if (pb != nullptr)
            {
                //如果队列是持久化的，那么就按照传递过来的属性决定消息是否持久化存储
                DeliveryMode mode=queue_is_durable?pb->deliver_mode():DeliveryMode::UNDURABLE;
                msp->mutable_payload()->mutable_properties()->set_id(pb->id());
                msp->mutable_payload()->mutable_properties()->set_deliver_mode(mode);
                msp->mutable_payload()->mutable_properties()->set_routing_key(pb->routing_key());
            }
            else
            {
                //如果队列时持久化的，消息也就持久化存储
                DeliveryMode mode=queue_is_durable?DeliveryMode::DURABLE:DeliveryMode::UNDURABLE;
                msp->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid()); // 生成随机ID
                msp->mutable_payload()->mutable_properties()->set_deliver_mode(mode);
                msp->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            std::unique_lock<std::mutex> lock(_mutex);
            // DLOG("发布消息，消息模块，msg_id为：%s\n",msp->payload().properties().id().c_str());
            // 2，如果这条消息需要持久化存储，则使用文件存储
            if (msp->payload().properties().deliver_mode() == DeliveryMode::DURABLE)
            {
                msp->mutable_payload()->set_valid(MESSAGE_VALID);
                bool ret = _mapper.insert(msp); // 将该消息写入到文件中
                if (ret == false)
                {
                    DLOG("消息写入文件失败");
                    return false;
                }
                _total_count += 1;
                _valid_count += 1;
                // 同时存储到持久化消息哈希表中
                _durable_msg.insert(std::make_pair(msp->payload().properties().id(), msp));
            }
            // 3,存储在内存中
            _send_msgs.push_back(msp);
            return true;
        }
        // 获取一条消息
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_send_msgs.size() == 0)
            {
                return MessagePtr();
            }
            MessagePtr msp = _send_msgs.front();
            _send_msgs.pop_front();
            // 将该消息对象向待确认哈希表中存储一份
            _waitack_msg.insert(std::make_pair(msp->payload().properties().id(), msp));
            return msp;
        }
        // 移除一条消息,判断是否需要进行垃圾回收
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            // 1,从待确认哈希表中查找该消息
            auto it = _waitack_msg.find(msg_id);
            if (it == _waitack_msg.end())
            {
                DLOG("没有找到要删除的消息%s\n", msg_id.c_str());
                return false;
            }
            // DLOG("确认一条线消息，从等待队列中移除\n");
            // 2,判断该消息是否持久化存储
            MessagePtr msp = it->second;
            if (msp->payload().properties().deliver_mode() == DeliveryMode::DURABLE)
            {
                // 3,从文件中删除该消息
                _mapper.remove(msp);
                _durable_msg.erase(msg_id); // 从持久化消息哈希表中也要是删除
                _valid_count -= 1;

                // 判断是否需要进行垃圾回收
                gc();
            }
            // 4,从内存中删除该徐消息
            _waitack_msg.erase(msg_id);
            return true;
        }
        // 等待确认的消息数量
        size_t waitack_count()
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            return _waitack_msg.size();
        }
        // 持久化消息的数量
        size_t durable_count()
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            return _durable_msg.size();
        }
        // 待推送消息数量
        size_t waitsend_count()
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            return _send_msgs.size();
        }
        size_t get_total_count()
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            return _total_count;
        }
        size_t get_valid_count()
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            return _valid_count;
        }
        void clear()
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _total_count = _valid_count = 0;
            _mapper.rmoveFile();
            _durable_msg.clear();
            _send_msgs.clear();
            _waitack_msg.clear();
        }

    private:
        // 检查是否需要进行垃圾回收
        bool gcCheck()
        {
            if (_total_count >= 2000 && _valid_count * 10 / _total_count <= 5)
            {
                return true;
            }
            return false;
        }
        void gc()
        {
            if (gcCheck() == false)
                return;

            std::list<MessagePtr> msgs = _mapper.gc();
            for (auto &msg : msgs)
            {
                auto it = _durable_msg.find(msg->payload().properties().id()); // 按理说这里的ID一定可以找到的
                if (it == _durable_msg.end())
                {
                    DLOG("文件中的消息在内存中不存在...\n");
                    continue;
                }
                // 更新每一条消息的实际存储位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            // 更新消息数量和有效消息数量
            _total_count = _valid_count = msgs.size();
        }

    private:
        std::mutex _mutex;
        std::string _qname;                                       // 队列名称
        MessageMapper _mapper;                                    // 实现消息的文件存储
        std::list<MessagePtr> _send_msgs;                         // 待推送消息链表
        std::unordered_map<std::string, MessagePtr> _durable_msg; // 持久化消息哈希表——消息ID和消息对象的映射
        std::unordered_map<std::string, MessagePtr> _waitack_msg; // 待确认消息哈希表——消息ID和消息对象的映射
        size_t _total_count;                                      // 消息总数量
        size_t _valid_count;                                      // 有效消息数量
    };

    // 消息的管理,以队列为单位
    // 每个队列有多个消息
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;
        MessageManager(std::string basedir)
            : _basedir(basedir)
        {
        }
        // 初始化一个消息队列
        void initQueueMessage(const std::string &qname)
        {
            // std::unique_lock<std::mutex> _lock(_mutex);
            // auto it = _queue_msgs.find(qname);
            // if (it != _queue_msgs.end())
            // {
            //     return;
            // }
            // //在构造一个消息队列时，需要进行恢复历史数据
            // //可能会很耗时，不建议在锁的保护下初始化
            // //其内部会有锁保护
            // QueueMessage::ptr qmp=std::make_shared<QueueMessage>(_basedir,qname);
            // _queue_msgs.insert(std::make_pair(qname,qmp));

            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end())
                {
                    return;
                }
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
            qmp->recovery();
        }
        // 删除一个消息队列
        void destoryQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("删除消息队列失败...\n");
                    return;
                }
                qmp = it->second;
                _queue_msgs.erase(it);
            }
            qmp->clear();
        }
        // 新增一条消息
        bool insert(const std::string &qname, const basicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("新增一条消息队列失败...\n");
                    return false;
                }
                qmp = it->second;
            }
            return qmp->insert(bp, body,queue_is_durable);
        }
        // 获取队首消息
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取一条消息队列失败...\n");
                    return MessagePtr();
                }
                qmp = it->second;
            }
            return qmp->front();
        }
        // 确认一条消息
        void ack(const std::string &qname, const std::string &msg_id)
        {
            // DLOG("在消息模块,msg_id为：%s",msg_id.c_str());
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("确认一条消息队列失败...\n");
                    return;
                }
                qmp = it->second;
            }
            qmp->remove(msg_id);
        }
        void clear()
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            for (auto &queue_msg : _queue_msgs)
            {
                queue_msg.second->clear();
            }
        }
        // 等待确认的消息数量
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列等待确认消息的数量失败...\n");
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }
        // 持久化消息的数量
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列持久化消息的数量失败...\n");
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }
        // 待推送消息数量
        size_t waitsend_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列待推送消息的数量失败...\n");
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitsend_count();
        }
        size_t get_total_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列等待确认消息的数量失败...\n");
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->get_total_count();
        }
        size_t get_valid_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列等待确认消息的数量失败...\n");
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->get_valid_count();
        }

    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs; // 队列名和队列消息对象的映射
    };
}
#endif